package contrab

import (
	"data_summary/common/config"
	db2 "data_summary/common/db"
	"data_summary/common/httpclient"
	"data_summary/common/models"
	rpcClient "data_summary/common/rpc/client"
	pb "data_summary/common/rpc/proto"
	logger "data_summary/logs_ope"
	"data_summary/service/dao"
	"fmt"
	"github.com/jakecoffman/cron"
	"github.com/ouqiang/goutil"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"
)

var (

	// 定时任务调度管理器
	serviceCron *cron.Cron
	// 同一任务是否有实例处于运行中
	runInstance Instance

	// 任务计数-正在运行的任务
	taskCount TaskCount

	// 并发队列, 限制同时运行的任务数量
	concurrencyQueue ConcurrencyQueue
)

type Task struct {
}

func NewTask() *Task {
	Cron := &Task{}
	return Cron
}

// 添加任务
func (task *Task) Add(taskModel models.Task) {
	if taskModel.Level == models.TaskLevelChild {
		logger.Errorf("添加任务失败#不允许添加子任务到调度器#任务Id-%d", taskModel.Id)
		return
	}
	taskFunc := task.createJob(taskModel)
	if taskFunc == nil {
		logger.Error("创建任务处理Job失败,不支持的任务协议#", taskModel.Protocol)
		return
	}

	cronName := strconv.Itoa(taskModel.Id)
	err := goutil.PanicToError(func() {
		serviceCron.AddFunc(taskModel.Spec, taskFunc, cronName)
	})
	if err != nil {
		logger.Error("添加任务到调度器失败#", err)
	}
}

func (task *Task) createJob(taskModel models.Task) cron.FuncJob {
	handler := getHandler(&taskModel)
	if handler == nil {
		return nil
	}
	taskFunc := func() {
		taskCount.Add()
		defer taskCount.Done()

		taskLogId := beforeExecJob(taskModel)
		if taskLogId <= 0 {
			return
		}
		if taskModel.Multi == 0 {
			runInstance.add(taskModel.Id)
			defer runInstance.done(taskModel.Id)
		}
		concurrencyQueue.Add()
		defer concurrencyQueue.Done()
		//fmt.Print("sdadsa23123131")
		logger.Info("开始执行任务#%s#命令-%s", taskModel.Name, taskModel.Command)
		taskResult := execJob(handler, taskModel, taskLogId)
		logger.Info("任务完成#%s#命令-%s", taskModel.Name, taskModel.Command)
		afterExecJob(taskModel, taskResult, taskLogId)
	}

	return taskFunc
}

// 直接运行任务
func (task *Task) Run(taskModel models.Task) {
	go task.createJob(taskModel)()
}

// 等待所有任务结束后退出
func (task *Task) WaitAndExit() {
	serviceCron.Stop()
	taskCount.Exit()
}

func getHandler(taskModel *models.Task) Handler {
	fmt.Print("Protocol--", taskModel.Protocol)
	var handler Handler = nil
	switch taskModel.Protocol {
	case models.TaskHTTP:
		handler = new(HTTPHandler)
	case models.TaskRPC:
		handler = new(RPCHandler)
	}
	return handler
}

// HTTP任务
type HTTPHandler struct{}

// http任务执行时间不超过300秒
const HttpExecTimeout = 300

func (h *HTTPHandler) Run(taskModel models.Task, taskUniqueId int64) (result string, err error) {
	if taskModel.Timeout <= 0 || taskModel.Timeout > HttpExecTimeout {
		taskModel.Timeout = HttpExecTimeout
	}
	var resp httpclient.ResponseWrapper
	if taskModel.HttpMethod == models.TaskHTTPMethodGet {
		resp = httpclient.Get(taskModel.Command, taskModel.Timeout)
	} else {
		urlFields := strings.Split(taskModel.Command, "?")
		taskModel.Command = urlFields[0]
		var params string
		if len(urlFields) >= 2 {
			params = urlFields[1]
		}
		resp = httpclient.PostParams(taskModel.Command, params, taskModel.Timeout)
	}
	// 返回状态码非200，均为失败
	if resp.StatusCode != http.StatusOK {
		return resp.Body, fmt.Errorf("HTTP状态码非200-->%d", resp.StatusCode)
	}

	return resp.Body, err
}

// RPC调用执行任务
type RPCHandler struct{}

func (h *RPCHandler) Run(taskModel models.Task, taskUniqueId int64) (result string, err error) {
	fmt.Print("sasdasdsa", taskModel)
	taskRequest := new(pb.TaskRequest)
	taskRequest.Timeout = int32(taskModel.Timeout)
	taskRequest.Command = taskModel.Command
	taskRequest.Id = taskUniqueId
	resultChan := make(chan TaskResult, len(taskModel.Hosts))
	for _, taskHost := range taskModel.Hosts {
		go func(th models.TaskHostDetail) {
			output, err := rpcClient.Exec(th.Name, th.Port, taskRequest)
			errorMessage := ""
			if err != nil {
				errorMessage = err.Error()
			}
			outputMessage := fmt.Sprintf("主机: [%s-%s:%d]\n%s\n%s\n\n",
				th.Alias, th.Name, th.Port, errorMessage, output,
			)
			resultChan <- TaskResult{Err: err, Result: outputMessage}
		}(taskHost)
	}

	var aggregationErr error = nil
	aggregationResult := ""
	for i := 0; i < len(taskModel.Hosts); i++ {
		taskResult := <-resultChan
		aggregationResult += taskResult.Result
		if taskResult.Err != nil {
			aggregationErr = taskResult.Err
		}
	}

	return aggregationResult, aggregationErr
}

// 执行具体任务
func execJob(handler Handler, taskModel models.Task, taskUniqueId int64) TaskResult {
	fmt.Print("ex-handler", handler)
	defer func() {
		if err := recover(); err != nil {
			logger.Error("panic#service/task.go:execJob#", err)
		}
	}()
	// 默认只运行任务一次
	var execTimes int8 = 1
	if taskModel.RetryTimes > 0 {
		execTimes += taskModel.RetryTimes
	}
	var i int8 = 0
	var output string
	var err error
	for i < execTimes {
		output, err = handler.Run(taskModel, taskUniqueId)
		if err == nil {
			return TaskResult{Result: output, Err: err, RetryTimes: i}
		}
		i++
		if i < execTimes {
			logger.Warnf("任务执行失败#任务id-%d#重试第%d次#输出-%s#错误-%s", taskModel.Id, i, output, err.Error())
			if taskModel.RetryInterval > 0 {
				time.Sleep(time.Duration(taskModel.RetryInterval) * time.Second)
			} else {
				// 默认重试间隔时间，每次递增1分钟
				time.Sleep(time.Duration(i) * time.Minute)
			}
		}
	}

	return TaskResult{Result: output, Err: err, RetryTimes: taskModel.RetryTimes}
}

// 任务前置操作
func beforeExecJob(taskModel models.Task) (taskLogId int64) {
	if taskModel.Multi == 0 && runInstance.has(taskModel.Id) {
		_, err := createTaskLog(taskModel, models.Cancel)
		if err != nil {
			return 0
		}
		return
	}
	taskLogId, err := createTaskLog(taskModel, models.Running)
	if err != nil {
		logger.Error("任务开始执行#写入任务日志失败-", err)
		return
	}
	logger.Info("任务命令-%s", taskModel.Command)

	return taskLogId
}

// 初始化任务, 从数据库取出所有任务, 添加到定时任务并运行
func (task *Task) Initialize() {
	serviceCron = cron.New()
	serviceCron.Start()

	concurrencyQueue = ConcurrencyQueue{queue: make(chan struct{}, config.ServerConfig.ConcurrencyQueue)}
	taskCount = TaskCount{sync.WaitGroup{}, make(chan struct{})}
	go taskCount.Wait()

	logger.Info("开始初始化定时任务")
	Db := db2.GetDb("database")
	taskModel := dao.NewTaskDao(Db)
	taskNum := 0
	page := 1
	pageSize := 1000
	maxPage := 1000
	for page < maxPage {
		taskList, err := taskModel.GetList(map[string]interface{}{"status": models.Enabled, "level": models.TaskLevelParent}, "id desc", page, pageSize)
		if err != nil {
			logger.Fatalf("定时任务初始化#获取任务列表错误: %s", err)
		}
		if len(taskList) == 0 {
			break
		}
		for _, item := range taskList {
			task.Add(item)
			taskNum++
		}
		page++
	}
	logger.Infof("定时任务初始化完成, 共%d个定时任务添加到调度器", taskNum)
}

// 任务执行后置操作
func afterExecJob(taskModel models.Task, taskResult TaskResult, taskLogId int64) {
	//_, err := updateTaskLog(taskLogId, taskResult)
	//if err != nil {
	//	logger.Error("任务结束#更新任务日志失败-", err)
	//}
	//
	//// 发送邮件
	//go SendNotification(taskModel, taskResult)
	//// 执行依赖任务
	//go execDependencyTask(taskModel, taskResult)
}

// 创建任务日志
func createTaskLog(taskModel models.Task, status models.Status) (int64, error) {
	Db := db2.GetDb("database")
	taskLogModel := models.TaskLog{}
	taskLogModel.TaskId = taskModel.Id
	taskLogModel.Name = taskModel.Name
	taskLogModel.Spec = taskModel.Spec
	taskLogModel.Protocol = taskModel.Protocol
	taskLogModel.Command = taskModel.Command
	taskLogModel.Timeout = taskModel.Timeout
	if taskModel.Protocol == models.TaskRPC {
		aggregationHost := ""
		for _, host := range taskModel.Hosts {
			aggregationHost += fmt.Sprintf("%s - %s<br>", host.Alias, host.Name)
		}
		taskLogModel.Hostname = aggregationHost
	}
	taskLogModel.StartTime = time.Now()
	taskLogModel.Status = status
	err := Db.Create(&taskLogModel).Error
	return taskLogModel.Id, err
}
